package cn.doitedu.kafka.rebalance;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class RebalanceTest {

    public static void main(String[] args) throws InterruptedException {
        Properties properties=new Properties();
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("bootstrap.servers","doit01:9092,doit02:9092,doit03:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"1");

        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"gp-1");
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);

        put(producer,"tpc-a","tpc-b","tpc-c");



    }


    public static void put(KafkaProducer<String, String> producer,String... tpc) throws InterruptedException {

        for(int i=0;i<100000;i++){
            Thread.sleep(1000);
            ProducerRecord<String, String> record = new ProducerRecord<>(tpc[i%3], tpc[i%3]+"-key-" + i, "value-" + i);
            producer.send(record);
        }
    }

}
